package RedisObject;

import Dict.Dict;
import IntSet.IntSet;
import QuickList.QuickList;
import RedisObject.enums.RedisEncoding;
import RedisObject.enums.RedisType;
import SDS.SDS;
import SkipList.SkipList;
import ZipList.ZipList;

import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.ObjectOutputStream;
import java.nio.charset.StandardCharsets;
import java.util.List;
import java.util.Map;
import java.util.Objects;

public class RedisFactory {

    // 创建字符串对象
    public static RedisObject createString(Object value) {
        RedisEncoding encoding;
        Object ptr;
        // 如果value是数字，直接将value赋值给ptr
        if (isInteger(value)) {
            encoding = RedisEncoding.REDIS_ENCODING_INT;
            ptr = value;
        }
        // 如果字符串的长度从<=44，将SDS拼接在RedisObject后面
        else if (((String) value).length() <= 44) {
            encoding = RedisEncoding.REDIS_ENCODING_EMBSTR;
            ptr = new SDS((String) value);
        }
        // 如果字符串的长度从>44，将ptr指向SDS
        else {
            encoding = RedisEncoding.REDIS_ENCODING_RAW;
            ptr = new SDS((String) value);
        }
        return new RedisObject(RedisType.REDIS_STRING, encoding, ptr);
    }

    private static boolean isInteger(Object value) {
        return value instanceof Integer;
    }

    // 创建列表对象
    public static RedisObject createList(List<Object> values) {
        RedisEncoding encoding;
        Object ptr;

        // 如果values的长度<=512且每个元素的长度<=64，用ZipList
        if (values.size() <= 512 && values.stream().allMatch(v -> getByteSize(v) <= 64)) {
            encoding = RedisEncoding.REDIS_ENCODING_ZIPLIST;
            ZipList zipList = new ZipList(values.size());
            values.forEach(v -> zipList.insert(getByte(v)));
            ptr = zipList;
        }

        // 如果values的长度>512或某个元素的长度>64，用QuickList
        else {
            encoding = RedisEncoding.REDIS_ENCODING_QUICKLIST;
            QuickList quickList = new QuickList(values.size());
            values.forEach(v -> quickList.insert(getByte(v)));
            ptr = quickList;
        }

        return new RedisObject(RedisType.REDIS_LIST, encoding, ptr);
    }

    private static byte[] getByte(Object obj) {
        if (obj instanceof String) {
            return ((String) obj).getBytes(StandardCharsets.UTF_8);
        } else {
            try (ByteArrayOutputStream ops = new ByteArrayOutputStream();
                 ObjectOutputStream oos = new ObjectOutputStream(ops)) {
                oos.writeObject(obj);
                oos.flush();
                return ops.toByteArray();
            } catch (IOException e) {
                throw new RuntimeException(e);
            }
        }
    }

    private static int getByteSize(Object obj) {
        return Objects.requireNonNull(getByte(obj)).length;
    }

    // 创建集合对象
    public static RedisObject createSet(List<Object> values) {
        RedisEncoding encoding;
        Object ptr;

        // 少量整数用IntSet
        if (values.size() <= 512 && values.stream().allMatch(RedisFactory::isInteger)) {
            encoding = RedisEncoding.REDIS_ENCODING_INTSET;
            IntSet intSet = new IntSet();
            values.forEach(v -> intSet.insert((Integer) v));
            ptr = intSet;
        }

        // 其他情况用Dict
        else {
            encoding = RedisEncoding.REDIS_ENCODING_HT;
            Dict dict = new Dict(values.size());
            values.forEach(v -> dict.insert((String) v, null));
            ptr = dict;
        }
        return new RedisObject(RedisType.REDIS_SET, encoding, ptr);
    }

    // 创建有序集合对象
    public static RedisObject createZSet(Map<Object, Double> values) {
        RedisEncoding encoding;
        Object ptr;
        // 少量数据用ZipList
        if (values.size() <= 128 && values.values().stream().allMatch(v -> getByteSize(v) <= 64)) {
            encoding = RedisEncoding.REDIS_ENCODING_ZIPLIST;
            ZipList zipList = new ZipList(values.size() * 2);
            values.forEach((k, v) -> {
                // 按照分数的大小进行排序
                zipList.insert(getByte(k), getByte(v));
            });
            ptr = zipList;
        }

        // 其他情况用SkipList
        else {
            encoding = RedisEncoding.REDIS_ENCODING_SKIPLIST;
            SkipList skipList = new SkipList();
            values.forEach(skipList::insert);
            ptr = skipList;
        }

        return new RedisObject(RedisType.REDIS_ZSET, encoding, ptr);
    }

    // 创建哈希对象
    public static RedisObject createHash(Map<Object, Object> values) {
        RedisEncoding encoding;
        Object ptr;

        // 少量数据用ZipList
        if (values.size() <= 512 && values.values().stream().allMatch(v -> getByteSize(v) <= 64)) {
            encoding = RedisEncoding.REDIS_ENCODING_ZIPLIST;
            ZipList zipList = new ZipList(values.size() * 2);

            // 将键值对插入ZipList（相邻）
            values.forEach((k, v) -> {
                zipList.insert(getByte(k));
                zipList.insert(getByte(v));
            });

            ptr = zipList;
        }

        // 大量数据用Dict
        else {
            encoding = RedisEncoding.REDIS_ENCODING_HT;
            Dict dict = new Dict(values.size());
            values.forEach((k, v) -> dict.insert((String) k, v));
            ptr = dict;
        }

        return new RedisObject(RedisType.REDIS_HASH, encoding, ptr);
    }
}
